package com.vip.iot.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.vip.iot.entity.IotAlarmCustom;
import com.vip.iot.entity.Vector3D;
import com.vip.iot.protocol.mqtt.IMqttInstance;
import com.vip.iot.protocol.mqtt.MqttInstanceImpl;
import com.vip.iot.util.CommonUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Iterator;
import java.util.UUID;

/**
 * 消息订阅Service
 *
 * @author wgb
 * @date 2020/6/18 9:39
 */
@Slf4j
@Component
public class MqttSubMsgService implements ApplicationRunner {
    /**
     * MQTT服务端 ip:port
     */
    @Value("${mqtt.server.url}")
    private String url;
    /**
     * MQTT登录账户名
     */
    @Value("${mqtt.server.username}")
    private String username;
    /**
     * 密码
     */
    @Value("${mqtt.server.password}")
    private String password;
    /**
     * 报警消息主题
     */
    @Value("${mqtt.topic.alarm}")
    private String alarmTopic;
    /**
     * 报警主题
     */
    @Value("${mqtt.topic.ring-location}")
    private String ringLocationTopic;
    /**
     * 基站的初始位置
     */
    @Value("${location.base.station.source}")
    private String stationLocationSource;
    @Autowired
    private KafkaTemplate kafkaTemplate;
    /**
     * 作为MQTT实例单例模式，提供单个MQTT连接
     */
    private static IMqttInstance client = null;

    /**
     * 单例模式，获取ImqttInstance实现类mqttInstanceImpl
     *
     * @return
     */
    public IMqttInstance getMqttInstance() {
        if (client == null) {
            client = new MqttInstanceImpl();
            client.init(url, username, password);
        }
        return client;
    }

    /**
     * 订阅报警消息
     */
    public void subscribeAlarmMessage() {
        IMqttInstance mqttInstance = getMqttInstance();
        try {
            mqttInstance.subscribe(alarmTopic, (t, m) -> {
                log.info("收到报警消息：{}", m);
//                String kafkaMessage = getAlarmInformToKafkaMessage(m, "alarm", "设备id", "1", "hg");
                sentMessageToKafka(t.replaceAll("/", ""), m);
            });
        } catch (MqttException e) {
            log.error("报警消息订阅失败！");
        }
        log.info("已订阅报警信息……");
    }

    /**
     * 订阅手环定位消息
     */
    public void subscribeHandleRingLocationMessage() {
        IMqttInstance mqttInstance = getMqttInstance();
        try {
            mqttInstance.subscribe(ringLocationTopic, (t, m) -> {
                log.info("收到手环定位消息：{}", m);
//                String kafkaMessage = getAlarmInformToKafkaMessage(m, "location", "设备ID", "1", "hg");
                sentMessageToKafka(t.replaceAll("/", ""), m);
            });
        } catch (MqttException e) {
            log.error("手环定位消息订阅失败！");
        }
        log.info("已订阅手环定位信息……");
    }

    /**
     * 将订阅到的消息转换为IotAlarmCustom
     *
     * @param message     MQTT源消息体
     * @param msgType     消息类型
     * @param id          设备ID
     * @param state       消息状态
     * @param companyCode 公司编码
     * @return 要推送到kafka服务端的消息体
     */
    public String getAlarmInformToKafkaMessage(String message, String msgType, String id, String state, String companyCode) {
        Object data = messageOfModelChangePos(message);
        IotAlarmCustom iotAlarmCustom = new IotAlarmCustom(msgType, id, state, data);
        iotAlarmCustom.setCompanyCode(companyCode);
        iotAlarmCustom.setMsgId(UUID.randomUUID().toString());
        return JSON.toJSONString(iotAlarmCustom);
    }

    /**
     * 将接收到的message中的pos值改为经纬度,并封装为对象，便于传给IotAlarmCustom对象中的data属性
     *
     * @param message MQTT源消息体
     * @return 源消息体的pos坐标属性值转换为经纬度后的Object对象
     */
    public Object messageOfModelChangePos(String message) {
        int start = message.indexOf("\"pos\":\"");
        int end = message.indexOf("\"", start + 7);
        String pos = message.substring(start + 7, end);
        String[] xyz = pos.split(",");
        Vector3D lonLatSourcePoint = locationSourceToVector3D(stationLocationSource);
        Vector3D mercatorSourcePoint = CommonUtils.lonLat2Mercator(lonLatSourcePoint);
        Vector3D mercator = new Vector3D(mercatorSourcePoint.getX() + Double.parseDouble(xyz[0]),
              mercatorSourcePoint.getY() + Double.parseDouble(xyz[1]), mercatorSourcePoint.getZ() + Double.parseDouble(xyz[2]));
        Vector3D lonlat = CommonUtils.Mercator2lonLat(mercator);
        String posData = lonlat.getX() + "," + lonlat.getY() + "," + lonlat.getZ();
        String entityStr = message.replaceAll(pos, posData);
        JSONObject jo = JSON.parseObject(entityStr);
        Iterator<String> keySet = jo.keySet().iterator();
        String type = keySet.next();
        return jo.getJSONArray(type).toJavaList(Object.class);
    }

    /**
     * 定位基站经纬度字符串转为Vector3D
     *
     * @param locationSourceLonLat 定位基站经纬度字符串
     * @return
     */
    public Vector3D locationSourceToVector3D(String locationSourceLonLat) {
        String[] xyz = locationSourceLonLat.split(",");
        return new Vector3D(Double.parseDouble(xyz[0]), Double.parseDouble(xyz[1]), Double.parseDouble(xyz[2]));
    }

    /**
     * 推送主题&数据到kafka服务端
     *
     * @param topic        要推送的主题
     * @param kafkaMessage 要推送的数据
     */
    public void sentMessageToKafka(String topic, String kafkaMessage) {
        ListenableFuture future = kafkaTemplate.send(topic, kafkaMessage);
//        future.addCallback(o -> log.info("消息推送到kafka成功!"),
        future.addCallback(o -> {},
              throwable -> log.error("消息推送kafka失败!"));
    }

    @Override
    public void run(ApplicationArguments args) {
        /** 订阅报警消息 */
        subscribeAlarmMessage();

        /** 订阅手环定位消息 */
        subscribeHandleRingLocationMessage();
    }
}
